/* Copyright  2009-2010
   @Author
   Richard Changde Yin            e-mail yinchangde@hotmail.com

   This program is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License as published by
   the Free Software Foundation; version 2 of the License.

   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU General Public License for more details.

   You should have received a copy of the GNU General Public License
   along with this program; if not, write to the Free Software
   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */

/**
 * @filename: database.cc
 */
#include <server_includes.h>

#include <databaseManagerImpl.h>
#include <session.h>
#include <table.h>
#include <current_session.h>

#include <mutex.h>
#include <database.h>

#include <os.h>
#include <allocator.h>
#include <catalogtables.h>
#include <transaction.h>
#include <lock.h>
#include <debug.h>
#include <read-config.h>
#include <process.h>

#include <gettext.h>
#include <error.h>
#include <errmsg_print.h>

extern "C" void free_tbl_ptr_struct(void *entry);

extern "C" unsigned char *tbl_ptr_struct_key(const unsigned char *buff,
                                              size_t *length,
                                              bool );
extern "C" void free_tbl_field (void* entry) ;

extern "C" unsigned char* get_tbl_field_key(const unsigned char* entry,
                                            size_t *length,bool);

using namespace BackEnd;

Database::~Database()
{
    if (isTblNameHashInited)
        hash_free(&tablename_hash) ;

    if (isTblFieldHashInited)
        hash_free(&tablefield_hash);
}
int Database::initAllocDatabaseMutex()
{
    static char _name[] = "allocdb";
    return metaData_->dbAllocMutex_.init(_name);
}
/*
 * SYNOPSIS initTblNameHash
 * @parameter  void
 * @return 0   on success
 *         1   on error
 */

int Database::initTblNameHash()
{
    errmsg_printf(ERRMSG_LVL_DBUG,_("Database::initialize table name hash.\n"));
    if(!hash_init(&tablename_hash,table_alias_charset,
                            (uint32_t) 1000, 0,
                            0,
                            tbl_ptr_struct_key,
                            free_tbl_ptr_struct,
                            HASH_UNIQUE))
    {
      isTblNameHashInited = true ;
      return (0) ;
    }
    isTblNameHashInited = false ;
    return (1) ;
}
/**
 * SYNOPSIS initTblFieldHash
 *
 * @param    void
 * @return   0  on success
 *           1  on error
 */

int Database::initTblFieldHash()
{
    errmsg_printf(ERRMSG_LVL_DBUG,_("Database::initialize table field hash.\n"));
    if(!hash_init(&tablefield_hash,system_charset_info,
                                 (uint32_t) 1000, 0,
                                  0,
                                  (hash_get_key)get_tbl_field_key,
                                  (hash_free_key)free_tbl_field,
                                  HASH_UNIQUE))
    {
      isTblFieldHashInited = true ;
      return (0) ;
    }
    isTblFieldHashInited = false ;
    return (1) ;
}
/**
 * SYNOPSIS:  insertTblNamePtr2Hash
 *
 * @param record
 * @return  1  On Error
 *          0  Successful
 */
int Database::insertTblNamePtr2Hash(unsigned char* record)
{
   if(isTblNameHashInited){
     if(hash_insert(&tablename_hash,record)) return 1 ;
       return (0) ;
   }
   return (1) ;
}
/**
 * SYNOPSIS  insertTblFieldPtr2Hash
 *
 * @param   record    unsigned char*
 * @return  1 On Error
 *          0 Successful
 */
int Database::insertTblFieldPtr2Hash (unsigned char* record)
{
    if(isTblFieldHashInited){
      if(hash_insert(&tablefield_hash,record)) return 1 ;
        return 0 ;
    }
   return 1 ;
}
/**
 * SYNOPSIS:  removeTblNamePtrFromHash
 *
 * @param key                    unsigned char*
 * @param keylength              uint32_t
 * @return   1 On Error
 *           0 Successful
 */

int Database::removeTblNamePtrFromHash(unsigned char* key,uint32_t keylength)
{
   if(isTblNameHashInited){
	   unsigned char* tblptr = 0;
     if ((tblptr = hash_search(&tablename_hash,key,keylength)))
     {
       if(hash_delete(&tablename_hash,tblptr)) return (1);
       return (0) ;
     }
   }
   return (1);
}
/*
 * SYNOPSIS  removeTblFieldPtrFromHash
 *
 * @param key
 * @param keylength
 * @return  1 On Error
 *          0 Successful
 */
int Database::removeTblFieldPtrFromHash(unsigned char* key,uint32_t keylength)
{
    if(isTblFieldHashInited){
        unsigned char* tblptr = 0;
        if ((tblptr = hash_search(&tablefield_hash,key,keylength)))
      {
        if(hash_delete(&tablefield_hash,tblptr)) return (1);
        return (0) ;
      }
    }
   return (1);
}
/*
 * SYNOPSIS:  searchTblNamePtr
 *
 * @param  key        unsigned char*
 * @param  keylength  uint32_t
 * @return      tblptr, if we found it from the hashtable
 *              NULL    Not found.
 */
struct tblname_ptr_struct* Database::searchTblNamePtr (unsigned char* key,
                                                       uint32_t keylength)
{
    if(isTblNameHashInited){
        unsigned char* tblptr = 0;
        tblptr = hash_search(&tablename_hash,key,(size_t)keylength);
        return (struct tblname_ptr_struct* )(tblptr) ;
    }
    return NULL ;
}
/**
 * SYNOPSIS: searchTblFieldPtr
 *
 * @param  key             unsigned char*
 * @param  keylength       uint32_t
 * @return tblptr   if we found from the hashtable
 *         NULL     if we did not find.
 */
Tbl_Field* Database::searchTblFieldPtr (unsigned char* key,uint32_t keylength)
{
    if(isTblFieldHashInited){
        unsigned char* tblptr = 0;
        tblptr = hash_search(&tablefield_hash,key,(size_t)keylength);
        return (Tbl_Field* )(tblptr) ;
    }
    return NULL ;
}

extern "C" unsigned char *tbl_ptr_struct_key(const unsigned char *buff,
                                             size_t *length,
                                             bool )
{
    struct tblname_ptr_struct * ptr = (struct tblname_ptr_struct*)buff ;
    *length = ptr->keylength ;
    return (unsigned char*) ptr->tblName_;

}

extern "C" void free_tbl_ptr_struct(void *entry)
{
    struct tblname_ptr_struct * ptr =(struct tblname_ptr_struct *)entry ;
    free((char*) ptr) ;
}

extern "C" void free_tbl_field (void* entry)
{
    Tbl_Field* tbl_field_ptr = (Tbl_Field*) entry;
    tbl_field_ptr->fldList.removeAll() ;  // we did not use the desctructor here
    free ((char*)entry);
}

extern "C" unsigned char* get_tbl_field_key(const unsigned char* entry,
                                            size_t *length,bool)
{
    Tbl_Field* tbl_field_ptr = (Tbl_Field*) entry;
    *length = strlen(tbl_field_ptr->add_str);
    return (unsigned char*)tbl_field_ptr->add_str;
}

BackEnd::DbRetVal Database::getAllocDatabaseMutex(bool procAccount)
{
    int _procslot = 0;
    if(current_session){_procslot=current_session->getProcSlot();}
    int ret= metaData_->dbAllocMutex_.getLock(_procslot, procAccount);
    if (ret) return (DbRetVal)HA_ERR_LOCK_WAIT_TIMEOUT;
    else return BackEnd::OK;
}
BackEnd::DbRetVal Database::releaseAllocDatabaseMutex(bool procAccount)
{
    int _procslot = 0;
    if(current_session){_procslot=current_session->getProcSlot();}
    metaData_->dbAllocMutex_.releaseLock(_procslot, procAccount);
    return BackEnd::OK;
}



int Database::initTransTableMutex()
{
    static char _name[]="transtable";
    return metaData_->dbTransTableMutex_.init(_name);
}
DbRetVal Database::getTransTableMutex()
{
    int _procslot = 0;
    if(current_session){_procslot=current_session->getProcSlot();}
    int ret = metaData_->dbTransTableMutex_.getLock(_procslot);
    if (ret) return (DbRetVal)HA_ERR_LOCK_WAIT_TIMEOUT;

    return BackEnd::OK;
}


DbRetVal Database::releaseTransTableMutex()
{
    int _procslot = 0;
    if(current_session){_procslot=current_session->getProcSlot();}
    metaData_->dbTransTableMutex_.releaseLock(_procslot);
    return BackEnd::OK;
}



int Database::initProcessTableMutex()
{
    char _name[] = "proctable";
    return metaData_->dbProcTableMutex_.init(_name);
}

DbRetVal Database::getProcessTableMutex(bool procAccount)
{
    int procslot = 0;
    if(current_session) procslot = current_session->getProcSlot();
    int ret = metaData_->dbProcTableMutex_.getLock(procslot, procAccount);
    if (ret) return (DbRetVal)(ret);
    else return BackEnd::OK;
}


DbRetVal Database::releaseProcessTableMutex(bool procAccount)
{
    int procslot = 0;
    if(current_session) procslot = current_session->getProcSlot();
    metaData_->dbProcTableMutex_.releaseLock(procslot, procAccount);
    return BackEnd::OK;
}

int Database::initDatabaseMutex()
{
    char _name[] = "db" ;
    return metaData_->dbMutex_.init(_name);
}
DbRetVal Database::getDatabaseMutex(bool procAccount)
{
    int _procslot = 0;
    if(current_session){_procslot=current_session->getProcSlot();}
    int ret = metaData_->dbMutex_.getLock(_procslot, procAccount);
    if (ret) return (DbRetVal)(ret);

    return BackEnd::OK;
}

DbRetVal Database::releaseDatabaseMutex(bool procAccount)
{
    int _procslot = 0;
    if(current_session){_procslot=current_session->getProcSlot();}
    metaData_->dbMutex_.releaseLock(_procslot, procAccount);
    return BackEnd::OK;
}

// Gets the free page
// Each page is segmented by PAGE_SIZE, so it checks the pageInfo
// of each page to determine if the page is free
// Algorithm is to scan through the pageInfo objects stored at
// address (db start address + i * PAGE_SIZE) where i = 1..n till end
// database
// But in case of large tuples, pages are merged, so there wont be
// PageInfo object on pages which are merged.
// These pages are skipped by checking the nextPageAfterMerge_ of PageInfo

//NOTE::IMPORTANT::assumes alloc database lock is taken before calling this
Page* Database::getFreePage()
{
    Page* page = getCurrentPage();
    //errmsg_printf(ERRMSG_LVL_DBUG, _("Database::getFreePage currentpage: %p\n"),page);

    PageInfo* pageInfo = ((PageInfo*)page);
    char* endAddr = ((char*)getMetaDataPtr())  + getMaxSize();
    int pageSize = PAGE_SIZE;
    bool isEndAddchk=false;
    while( 1 == pageInfo->isUsed_)
    {
        //If any pages are merged to store data larger than PAGE_SIZE
        //move to the next page after the merge and check whether it is used
        if ( pageInfo->nextPageAfterMerge_ == NULL) {
          pageInfo = (PageInfo*)((char*)pageInfo + pageSize);
        }
        else {
          pageInfo = (PageInfo*)pageInfo->nextPageAfterMerge_;
        }
        if((((char*) pageInfo) + pageSize) >= endAddr )
        {
          if(!isEndAddchk)
          {
            isEndAddchk=true;
            pageInfo=(PageInfo *)getFirstPage();
          }else break;
        }
        if ((char*)pageInfo >= endAddr)
        {
          errmsg_printf(ERRMSG_LVL_ERROR,
                       _("Error!Database::getFreePage returning invalid address: %p\n"),
                         pageInfo);
          return NULL;
        }
    }
    if (!isValidAddress(((char*) pageInfo) + pageSize))
    {
      errmsg_printf(ERRMSG_LVL_ERROR,
                    _("Error!Database::getFreePage returning invalid address: %p\n"),
                      pageInfo);
      return NULL;
    }
    setCurrentPage((Page*) pageInfo);
    //errmsg_printf(ERRMSG_LVL_DBUG, _("Database::getFreePage returning page address: %p\n"),
    //               pageInfo);
    return (Page*) pageInfo ;
}

//Used by large page which is greater than PAGE_SIZE
//NOTE::IMPORTANT::assumes alloc database lock is taken before calling this
Page* Database::getFreePage(size_t size)
{
    Page* page = getFirstPage();
    PageInfo* pageInfo = ((PageInfo*)page);
    uint32_t multiple = size / PAGE_SIZE;
    uint64_t offset = ((multiple + 1) * PAGE_SIZE);

    errmsg_printf(ERRMSG_LVL_DBUG, _("Database::getFreePage firstPage: %p size: %ld\n"),
                  page, size);
    char* endAddr = ((char*)getMetaDataPtr())  + getMaxSize();
    int pageSize = PAGE_SIZE;
    while(true){
         while( 1 == pageInfo->isUsed_)
        {
            //If any pages are merged to store data larger than PAGE_SIZE
            //move to the next page after the merge and check whether it is used
          if ( pageInfo->nextPageAfterMerge_ == NULL) {
            pageInfo = (PageInfo*)((char*)pageInfo + pageSize);
          }
          else {
            pageInfo = (PageInfo*)pageInfo->nextPageAfterMerge_;
          }
        }
        uint64_t i = 0;
        PageInfo *pInfo = pageInfo;
        if ((((char*)pInfo) + offset) >= endAddr)
        {
          errmsg_printf(ERRMSG_LVL_ERROR,
                        _("Error!Database::getFreePage returning invalid address: %p\n"),
                        ((char*)pInfo) + offset);
            return NULL;
        }
        for (i = 0; i< multiple + 1; i++)
        {
          if (1 == pInfo->isUsed_) break;
          pInfo = (PageInfo*)((char*)pInfo + pageSize);
        }
        if ( i == (multiple + 1))  break;
          pageInfo = (PageInfo*)((char*)pInfo + pageSize);
    }
    errmsg_printf(ERRMSG_LVL_DBUG, _("Database::getFreePage returning page address: %p\n"),
                       pageInfo);
    setCurrentPage((Page*) pageInfo);
    return (Page*) pageInfo ;
}

void Database::printStatistics()
{
    Page* page = getFirstPage();
    PageInfo* pageInfo = ((PageInfo*)page);
    int usedPageCount =0, usedMergedPageCount =0, totalPages=0;
    printf("<DatabaseStatistics>\n");
    printf("  <Database Name>  %s </Database Name>\n", getName());
    printf("  <Max Size> %ld </Max Size>\n", getMaxSize());
    printf("  <First Page> %p </First Page>\n", getFirstPage());
    while(isValidAddress((char*) pageInfo))
    {
        if (pageInfo == NULL) break;
        if (1 == pageInfo->isUsed_) {
           if ( pageInfo->nextPageAfterMerge_ == NULL) {
              pageInfo = (PageInfo*)((char*)pageInfo + PAGE_SIZE);
              usedPageCount++; totalPages++;
              continue;
           }
           else {
              pageInfo = (PageInfo*)pageInfo->nextPageAfterMerge_;
              usedMergedPageCount++; totalPages++;
              continue;
           }
        }
        pageInfo = (PageInfo*)((char*)pageInfo + PAGE_SIZE);
        totalPages++;
    }
    printf("  <Total Pages> %d </Total Pages>\n", totalPages);
    printf("  <Used Normal Pages> %d </Used Normal Pages>\n", usedPageCount);
    printf("  <Used Merged Pages> %d </Used Merged Pages>\n", usedMergedPageCount);
    printf("  <Chunks Used> %d </Chunks Used>\n", getNoOfChunks());
    printf("</DatabaseStatistics>\n");

    return ;
}


//called only in case of system database to create and initialize the chunk
//information
BackEnd::DbRetVal Database::createSystemDatabaseChunk(AllocType type,
                                                      size_t size,
                                                      int id)
{
    Chunk *chunk;
    if (id <0 )
    {
      return (DbRetVal)HA_ERR_SYS_INTERNAL;
    }
    chunk = getSystemDatabaseChunk(id);

    chunk-> setChunkNameForSystemDB(id);

    if (FixedSizeAllocator == type) chunk->setSize(size);
    //getDatabaseMutex();
    if (chunk->allocSize_ > PAGE_SIZE)
        chunk->curPage_ = getFreePage(chunk->allocSize_);
    else
        chunk->curPage_ = getFreePage();
    if ( chunk->curPage_ == NULL)
    {
       errmsg_printf(ERRMSG_LVL_ERROR,
                     _("Error!No free page is found during creating system database chunk.\n"));

       return (BackEnd::DbRetVal)HA_ERR_OUT_OF_MEM;
    }

    chunk->firstPage_ = chunk->curPage_;
    PageInfo* firstPageInfo = ((PageInfo*)chunk->firstPage_);
    firstPageInfo->setFirstPageAsUsed();
    chunk->setChunkID(id);
    chunk->setAllocType(type);

    if (chunk->allocSize_ > PAGE_SIZE)
    {
        uint32_t multiple = OS::floor(chunk->allocSize_ / PAGE_SIZE);
        uint64_t offset = ((multiple + 1) * PAGE_SIZE);
        firstPageInfo->nextPageAfterMerge_ = ((char*)firstPageInfo)+ offset;
        firstPageInfo->isMerged_   = true ;
    }

    if (0 == size)
    {
        VarSizeInfo *varInfo =
        (VarSizeInfo*)(((char*)firstPageInfo) + sizeof(PageInfo));
        varInfo->isUsed_ = 0;
        varInfo->size_ = PAGE_SIZE - sizeof(PageInfo) - sizeof(VarSizeInfo);

    }
    incrementChunk();
    return BackEnd::OK;
}

//This is never called currently. If situation arises will be coded later.
DbRetVal Database::deleteSystemDatabaseChunk(int id)
{

    Chunk *chunk = getSystemDatabaseChunk(id);
    chunk->setChunkID(-1);
    chunk->setSize(0);
    chunk->setAllocType(UnknownAllocator);
    //TODO::
    //chunk->pageList_
    //walk though the pageList ptr and get all the page pointers
    //then free all the pages used to store this by setting the
    //start of page to not used
    chunk->firstPage_ = NULL;
    chunk->curPage_ = NULL;
    decrementChunk();
    return BackEnd::OK;
}


void Database::createAllCatalogTables()
{
    //These are special chunks which hold catalog tables and other information
    //
    // chunk id 0 ->userChunkTable
    // chunk id 1 ->lockBucketHash
    // chunk id 2 ->lockTable
    //
    // chunk id 10->DATABASE
    // chunk id 11->USER
    // chunk id 12->CTABLE
    // chunk id 13->CFIELD
    // chunk id 14->CACCESS
    errmsg_printf(ERRMSG_LVL_INFO,
                _("Start to create system tables.\n"));
    createSystemTables();
    errmsg_printf(ERRMSG_LVL_INFO,
                _("Start to create Meta data tables.\n"));
    createMetaDataTables();
}
void Database::createSystemTables()
{
    createSystemDatabaseChunk(FixedSizeAllocator,
                                  sizeof(Chunk),
                                  UserChunkTableId);
    createSystemDatabaseChunk(FixedSizeAllocator,
                                  sizeof(Chunk),
                                  UserIndexChunkId);

    createSystemDatabaseChunk(FixedSizeAllocator,
                                  sizeof(Bucket)* LOCK_BUCKET_SIZE,
                                  LockTableHashBucketId);
    createSystemDatabaseChunk(FixedSizeAllocator,
                                  sizeof(Mutex)* LOCK_BUCKET_SIZE,
                                  LockTableMutexId);
    createSystemDatabaseChunk(FixedSizeAllocator,sizeof(LockHashNode),
                                  LockTableId);
    createSystemDatabaseChunk(FixedSizeAllocator,sizeof(TransHashNode),
                                   TransHasTableId);

    createSystemDatabaseChunk(VariableSizeAllocator,
                                  0, UndoLogTableID);
}
void Database::createMetaDataTables()
{
    createSystemDatabaseChunk(FixedSizeAllocator,
                                  sizeof(CDATABASEFILE), DatabaseTableId);
    /*
    createSystemDatabaseChunk(FixedSizeAllocator,
                                  sizeof(CUSER),     UserTableId);
    */
    createSystemDatabaseChunk(FixedSizeAllocator,
                                  sizeof(CTABLE),      TableTableId);
    createSystemDatabaseChunk(FixedSizeAllocator,
                                  sizeof(CFIELD),      FieldTableId);
    createSystemDatabaseChunk(FixedSizeAllocator,
                                  sizeof(CACCESS),     AccessTableId);
    createSystemDatabaseChunk(FixedSizeAllocator,
                                  sizeof(CINDEX),      IndexTableId);
    createSystemDatabaseChunk(FixedSizeAllocator,
                                  sizeof(CINDEXFIELD), IndexFieldTableId);
}

//used in case of system database
ThreadInfo* Database::getThreadInfo(int slot)
{
    static size_t offset = OS::alignLong(sizeof (DatabaseMetaData))     +
                           OS::alignLong( MAX_CHUNKS  * sizeof (Chunk)) +
                           OS::alignLong( (size_t)Conf::config.getMaxProcs()
                                          *sizeof(Transaction));
    size_t _slot = (size_t)slot;
    size_t off = offset + _slot * sizeof (ThreadInfo);
    return (ThreadInfo*)(((char*) metaData_) +  off);

}

//should be called only on system database
void* Database::allocLockHashBuckets()
{
    Chunk *chunk = getSystemDatabaseChunk(LockTableHashBucketId);
    void *ptr = chunk->allocate(this);
    if(ptr){
      ChunkIterator iter = chunk->getIterator();
      Bucket* buckets = (Bucket*)iter.nextElement();
      Bucket* bucket;
      for (int i=0; i< LOCK_BUCKET_SIZE; i++)
      {
        bucket = &(buckets[i]);
        bucket->rwlock = PTHREAD_RWLOCK_INITIALIZER;
        bucket->mutex_.init((char*)"LocktableHash");
        bucket->element_chunk = 0;
      }
    }
    return ptr;
}

int Database::initLockHashBucketTrees(DatabaseManagerImpl& dbmgr)
{
    Chunk *tChunk = getSystemDatabaseChunk(LockTableHashBucketId);
    ChunkIterator iter = tChunk->getIterator();

    Bucket* buckets = (Bucket*)iter.nextElement();
    Bucket* bucket;
    for (int i=0; i< LOCK_BUCKET_SIZE; i++)
    {
      bucket = &(buckets[i]);
      bucket->element_chunk = dbmgr.createUserIndexChunk(sizeof(BUCKET_TREE_ELEMENT));
      BKT_TREE_ARG* thearg = new BKT_TREE_ARG;
      thearg->dbptr= dbmgr.db();
      thearg->chunkptr=bucket->element_chunk ;
      thearg->alloc_type = assignment ;
      bucket->subtree.init_tree((unsigned)0,(int)sizeof(BUCKET_TREE_ELEMENT),
                              &Database::pointer_keys_compare,(void*)thearg);
    }
    return (0);
}

Bucket* Database::getLockHashBuckets()
{
    Chunk *tChunk = getSystemDatabaseChunk(LockTableHashBucketId);

    ChunkIterator iter = tChunk->getIterator();

    return (Bucket*)iter.nextElement();
}

DbRetVal Database::recoverMutex(Mutex* _mutex)
{
    //TODO: operations need to be undone before recovering the mutex.
    _mutex->recoverMutex();
    return OK ;
}
